package com.zetyun.tiger.connector.jdbc.source.action;

import com.zetyun.rt.meta.annotation.ActionMeta;
import com.zetyun.rt.sdk.internal.DelegateSourceAction;
import com.zetyun.rt.sdk.operator.OperatorContext;
import com.zetyun.tiger.connector.jdbc.source.JdbcSourceFunction;
import lombok.extern.slf4j.Slf4j;
import lombok.val;

@ActionMeta(
        id = "tiger.jdbc-streaming-source",
        name = "JDBC Streaming Source",
        tags = {"jdbc", "streaming"},
        category = "source",
        parameterClass = JdbcSourceParameter.class,
        description = "参考自 https://github.com/deep-bi/flink-connector-jdbc-source "

)
@Slf4j
public class JdbcSourceAction extends DelegateSourceAction {

    @Override
    public void init(OperatorContext context) {
        super.init(context);
        JdbcSourceParameter param = context.getParameters();
        log.info("{}  received param: {}", getClass().getSimpleName(), param);

        val jdbcSourceFun = new JdbcSourceFunction<>(param.buildJdbcSourceConfig());

        try {
            delegate(jdbcSourceFun);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }
}
